package mylab.starters.mq.impl.rabbitmq;


import com.rabbitmq.client.*;
import mylab.starters.mq.api.Message;
import mylab.starters.mq.api.MessageListener;
import mylab.starters.mq.api.Properties;
import mylab.starters.mq.config.MqConst;
import mylab.starters.mq.impl.AbstractConsumer;
import mylab.utils.spring.SpringUtil;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class RabbitConsumer extends AbstractConsumer {

    ConnectionFactory factory;
    Connection connection;
    Channel channel;
    String consumerTag;

    public RabbitConsumer(Properties properties, String groupId, String clientId) {
        super(properties, groupId, clientId);
        factory = SpringUtil.getBean(ConnectionFactory.class);
    }

    @Override
    protected void onSubscribe(String topic, MessageListener messageListener, String... tags) throws Exception {
        connection = factory.newConnection();
        channel = connection.createChannel();

        //创建队列
        channel.queueDeclare(topic, false, false, false, null);
        channel.queueBind(topic, MqConst.RABBIT_EXCHANGE, topic);
        channel.basicQos(1);

        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String content = new String(body, StandardCharsets.UTF_8);
                Message message = new Message(topic, content);
                messageListener.onSuccess(message);
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        consumerTag = channel.basicConsume(topic, false, consumer);
    }

    @Override
    protected void onUnsubscribe() throws Exception {
        channel.basicCancel(consumerTag);
        channel.close();
        connection.close();
    }

}
